package com.ruoyi.eqmonitor.utils;

import cn.hutool.core.codec.Base64;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.*;
import cn.hutool.json.JSONUtil;
import com.ruoyi.eqmonitor.annotation.EncodeBase64;
import com.ruoyi.eqmonitor.dispose.domain.OpentsdbTags;
import com.ruoyi.eqmonitor.exception.OpentsdbException;
import com.starwsn.protocol.json.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.nio.reactor.IOReactorException;
import org.opentsdb.client.OpenTSDBClient;
import org.opentsdb.client.OpenTSDBClientFactory;
import org.opentsdb.client.OpenTSDBConfig;
import org.opentsdb.client.bean.request.Point;
import org.opentsdb.client.bean.request.Query;
import org.opentsdb.client.bean.request.SubQuery;
import org.opentsdb.client.bean.request.SubQuery.Filter;
import org.opentsdb.client.bean.request.SubQuery.Filter.FilterType;
import org.opentsdb.client.bean.response.QueryResult;
import org.opentsdb.client.common.Json;
import org.opentsdb.client.exception.http.HttpException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Component;

import java.beans.IntrospectionException;
import java.beans.PropertyDescriptor;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * opentsdb工具类
 *
 * @author dh
 */
@Slf4j
@Component
public class OpenTSDBUtils {
    public static String OPENTSDB_TIMEZONE = "Asia/Shanghai";

    @Value("${opentsdb.host}")
    private String host;
    @Value("${opentsdb.port}")
    private Integer port;
    @Value("${opentsdb.metric}")
    private String metric;
    /** 超时时间 秒 */
    @Value("${opentsdb.timeout:30}")
    private int timeout;

    public String getHost() {
        return host;
    }

    public Integer getPort() {
        return port;
    }

    public String getMetric() {
        return metric;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setHost(String host) {
		this.host = host;
	}

	public void setPort(Integer port) {
		this.port = port;
	}

	public void setMetric(String metric) {
		this.metric = metric;
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	/**
     * 写数据使用client
     * @return
     * @throws IOReactorException
     */
    public OpenTSDBClient writeClient() throws IOReactorException {
        OpenTSDBConfig config = OpenTSDBConfig.address(host, port).httpConnectTimeout(timeout).config();
        return client(config);
    }

    /**
     * 同步写入数据
     */
    public boolean writeSync(List<Point> points) throws OpentsdbException {
        OpenTSDBClient client;
        try {
            client = writeClient();
        } catch (IOReactorException e) {
            throw new OpentsdbException("向opentsdb发起连接失败", e);
        }
        try {
            long start = System.currentTimeMillis();
            log.info("opentsdb执行存储 point信息:{}", JSON.toJSONString(points));
            client.putSyncWithCallBack(points);
            log.info("opentsdb执行存储成功, 耗时: {}", (System.currentTimeMillis() - start));
            return true;
        } catch (IOException | ExecutionException | InterruptedException | HttpException e) {
            throw new OpentsdbException("opentsdb执行存储失败", e);
        } finally {
            if (client != null) {
                // 关闭连接
                try {
                    client.gracefulClose();
                } catch (IOException ignored) {
                }
            }
        }
    }
    /**
     * 同步写入数据 增加版本号日志 -- 默认重试3次
     */
    public void writeSync(List<Point> points, String version) throws OpentsdbException {
        writeSync(points, version, 3);
    }

    /**
     * 同步写入数据 增加版本号日志
     */
    public void writeSync(List<Point> points, String version, Integer retryCount) throws OpentsdbException {
        OpenTSDBClient client;
        int count = 0;
        long start = System.currentTimeMillis();
        log.info("{} opentsdb执行存储 point信息:{}", version, JSONUtil.toJsonStr(points));
        // 执行并重试
        do {
            // 连接
            try {
                client = writeClient();
            } catch (Exception e) {
                if (count <= retryCount) {
                    ++count;
                    // 其他异常都执行重试
                    log.warn("第{}次尝试连接opentsdb失败: ", count);
                    try {
                        // 第一次失败 等待1秒, 第二次失败 等待2秒 以此类推
                        TimeUnit.SECONDS.sleep(count * 2L);
                    } catch (InterruptedException ex) {
                        throw new OpentsdbException(ex.getMessage());
                    }
                    continue;
                } else {
                    throw new OpentsdbException("向opentsdb发起连接失败", e);
                }
            }
            try {
                client.putSyncWithCallBack(points);
                log.info("{} opentsdb执行存储成功, 耗时: {}", version, (System.currentTimeMillis() - start));
                return;
            } catch (Exception e) {
                if (count <= retryCount) {
                    ++count;
                    // 其他异常都执行重试
                    log.warn("{} 第{}次尝试向opentsdb存储失败: {}", version, count, e.getMessage());
                    try {
                        // 第一次失败 等待1秒, 第二次失败 等待2秒 以此类推
                        TimeUnit.SECONDS.sleep(count);
                    } catch (InterruptedException ex) {
                        throw new OpentsdbException(ex.getMessage());
                    }
                } else {
                    throw new OpentsdbException("重试后opentsdb执行存储失败", e);
                }
            } finally {
                if (client != null) {
                    // 关闭连接
                    try {
                        client.gracefulClose();
                    } catch (IOException ignored) {
                    }
                }
            }
        } while (true);
    }

    /**
     * 同步删除
     * @param query 查询条件
     * @throws OpentsdbException 删除异常
     */
    public void deleteSync(Query query) throws OpentsdbException {
        OpenTSDBClient client = null;
        // 连接
        try {
            client = writeClient();
        } catch (Exception e) {
            throw new OpentsdbException("向opentsdb发起连接失败", e);
        }
        try {
            List<QueryResult> deleteResult = client.delete(query);
            log.info("opentsdb执行删除成功\nquery: {}\ndeleteResult:{}", query, deleteResult);
        } catch (Exception e) {
            log.error("opentsdb执行删除失败", e);
        } finally {
            if (client != null) {
                // 关闭连接
                try {
                    client.gracefulClose();
                } catch (IOException ignored) {
                }
            }
        }
    }

    /**
     * 查询使用client
     * @return
     * @throws IOReactorException
     */
    public OpenTSDBClient readClient() throws IOReactorException {
        OpenTSDBConfig.Builder builder = OpenTSDBConfig.address(host, port);
        builder.readonly().httpConnectTimeout(timeout).config();
        return client(builder.config());
    }

    public void closeClient(OpenTSDBClient client) {
        if (client != null) {
            //关闭连接
            try {
                client.gracefulClose();
            } catch (IOException e) {
                log.error("关闭opentsdb连接失败", e);
            }
        }
    }

    /**
     * 自定义config生成客户端方法
     * @param config 调用配置OpenTSDBConfig
     * @return
     * @throws IOReactorException
     */
    public OpenTSDBClient client(OpenTSDBConfig config) throws IOReactorException {
        return OpenTSDBClientFactory.connect(config);
    }

    /**
     * 自定义参数设置
     * 设置参数详情查看 {@link OpenTSDBConfig.Builder}
     * 设置参数完成后调用config()方法生成OpentsdbConfig
     * @return OpenTSDBConfig.Builder
     */
    public OpenTSDBConfig.Builder customerConfig() {
        return OpenTSDBConfig.address(host, port);
    }

    /**
     * 生成查询过滤器
     * @param filterType 过滤器类型 {@link FilterType}
     * @param tagk 过滤的标签k
     * @param filter 过滤的标签k的值
     * @return 查询过滤器
     */
    public static Filter createFilter(FilterType filterType, String tagk, String filter) {
        return createFilter(filterType, tagk, filter, false);
    }

    /**
     * 生成查询过滤器
     * @param filterType 过滤器类型 {@link FilterType}
     * @param tagk 过滤的标签k
     * @param filter 过滤的标签k的值
     * @param groupBy 是否分组
     * @return 查询过滤器
     */
    public static Filter createFilter(FilterType filterType, String tagk, String filter, Boolean groupBy) {
        return Filter
                .filter(filterType, tagk, filter, groupBy)
                .build();
    }

    /**
     * 序列化opentsdb时间
     * @param strDate 字符串序列化时间
     * @return opentsdb查询所有时间戳单位到秒
     */
    public static Long serializedTime(String strDate) {
        return DateUtil.parse(strDate).getTime() / 1000;
    }

    /**
     * 序列化opentsdb时间
     * @param date 时间
     * @return opentsdb查询所有时间戳单位到秒
     */
    public static Long serializedTime(Date date) {
        return date.getTime() / 1000;
    }

    /**
     * 将年月序列化opentsdb时间
     * @param strDate 字符串序列化时间
     * @return opentsdb查询所有时间戳单位到秒
     */
    public static Long serializedTimeMonth(String strDate) {
        return DateUtil.parse( strDate, "yyyy-MM").getTime() / 1000;
    }

    /**
     * 反序列化opentsdb时序日期单位
     * @param timestamp opentsdb时序时间
     * @return 反序列化日期单位
     */
    public static String deserializeDate(Long timestamp) {
        return DateUtil.format(new Date(timestamp * 1000), "yyyy-MM-dd");
    }

    /**
     * 反序列化opentsdb时序时间单位 不含秒
     * @param timestamp opentsdb时序时间
     * @return 反序列化时间单位
     */
    public static String deserializeTime(Long timestamp) {
        return DateUtil.format(new Date(timestamp * 1000), "HH:mm");
    }

    /**
     * 反序列化opentsdb时序日期时间单位 'yyyy-MM-dd HH:mm:ss'
     * @param timestamp opentsdb时序时间
     * @return 反序列化时间单位
     */
    public static String deserialize(Long timestamp) {
        return DateUtil.formatDateTime(new Date(timestamp * 1000));
    }

    /**
     * 根据时间偏移到当日时间最后一秒
     * @param strDate 字符串日期
     * @return 偏移后时间戳 单位到秒
     */
    public static Long lastTimeOffsetOfTheDay(String strDate) {
        DateTime dateTime = DateUtil.parseDate(strDate);

        return dateTime.offset(DateField.DAY_OF_MONTH, 1).offset(DateField.SECOND, -1).getTime() / 1000;
    }

    /**
     * 根据月偏移到当月最后一天时间的最后一秒
     * @param strDate 字符串日期
     * @return 偏移后时间戳 单位到秒
     */
    public static Long lastTimeOffsetOfTheDayByMonth(String strDate) {
        DateTime dateTime = DateUtil.parse(strDate, "yyyy-MM");

        return dateTime.offset(DateField.MONTH, 1).offset(DateField.SECOND, -1).getTime() / 1000;
    }

    /**
     * opentsdb默认数据转换echarts数据
     * @param queryResults opentsdb结果集
     * @param callback 转换函数
     * @return echarts结果集
     */
    public static List<Map<String, String>> convertToChart(List<QueryResult> queryResults, Callback callback) {
        List<Map<String, String>> list = Collections.synchronizedList(new ArrayList<>());
        queryResults.parallelStream().forEach(result -> {
            for (Map.Entry<Long, Number> dpsEntry : result.getDps().entrySet()) {
                Map<String, String> map = callback.create(dpsEntry, result, null);
                list.add(map);
            }
        });
        return list;
    }

    /**
     * echarts样例
     *  "date": "2020-11-26",
     *       "0x05D1": 0.08,
     *       "factory": "排水户",
     *       "factoryId": "0D8HT0",
     *       "robName": "在线液位计（窨井）23",
     *       "time": "12:00"
     * opentsdb分组后数据转换echarts数据
     * @param groupByResult opentsdb返回数据分组后集合 steam lambda表达式 Collectors.groupingBy
     * @param callback 转换函数
     * @return echats结果集
     */
    public static List<Map<String, String>> convertToChart(Map<String, List<QueryResult>> groupByResult, Callback callback) {

        List<Map<String, String>> results = Collections.synchronizedList(new ArrayList<>());
        // 处理返回结果 转化echarts需要的格式
        groupByResult.entrySet().parallelStream().forEach(e -> {
            List<QueryResult> snCodeResults = e.getValue();
            if (snCodeResults.isEmpty()) return;
            // 获取当前sncode第一条时序数据集，按照时许集作为参考查询当前设备其它监测项
            QueryResult queryResult = snCodeResults.get(0);
            Set<Map.Entry<Long, Number>> entries = queryResult.getDps().entrySet();
            // 遍历时序集
            for (Iterator<Map.Entry<Long, Number>> iterator = entries.iterator(); iterator.hasNext(); ) {
                Map.Entry<Long, Number> dataPoint = iterator.next();
                // 按时序合并其它监测项相同时序为一条结果
                Map<String, String> map = callback.create(dataPoint, queryResult, snCodeResults);
                // 存放echarts结果
                results.add(map);
            }
        });
        return results;
    }

    /**
     * 转换回调，自定义数据内部属性
     */
    public interface Callback {
        /**
         * 创建echarts单条数据
         * @param dataPoint 当前时序 时间：值
         * @param queryResult 当前查询结果 包含时序集
         * @param QueryResults opentsdb 查询结果集
         * @return
         */
        Map<String, String> create(Map.Entry<Long, Number> dataPoint, QueryResult queryResult, List<QueryResult> QueryResults);
    }

    /**
     * 转换为opentsdb需要的数据结构
     * @param time 时序序列(时间戳)
     * @param value 时序数值
     */
    public Point buildPoint(long time, Number value, OpentsdbTags opentsdbTags) throws IllegalAccessException {
        return Point.metric(this.metric).tag(encodeToMap(opentsdbTags)).value(time, NumberUtil.round(value.doubleValue(), 3)).build();
    }


    /**
     * 转换为opentsdb需要的数据结构
     * @param date 时间
     * @param value 时序数值
     * @param opentsdbTags 数据tags
     */
    public Point buildPoint(Date date, Number value, OpentsdbTags opentsdbTags) throws IllegalAccessException {
        Long serializedTime = OpenTSDBUtils.serializedTime(date);
        return buildPoint(serializedTime, value, opentsdbTags);
    }

    /**
     * 转换为opentsdb需要的数据结构
     * @param metric metric
     * @param time 时序序列(时间戳)
     * @param value 时序数值
     */
    public static Point buildPoint(String metric, long time, Number value, OpentsdbTags opentsdbTags) throws IllegalAccessException {
        return Point.metric(metric).tag(encodeToMap(opentsdbTags)).value(time, NumberUtil.round(value.doubleValue(), 3)).build();
    }

    /**
     * 编码并转换为Map
     * @return map
     */
    public static Map<String, String> encodeToMap(OpentsdbTags opentsdbTags) throws IllegalAccessException {
        if (opentsdbTags == null) {
            throw new RuntimeException("需要转换的对象不允许为空");
        }
        Map<String, String> mapping = new HashMap<>();
        Field[] fields = ReflectUtil.getFields(opentsdbTags.getClass());
        Field.setAccessible(fields, true);
        for (Field field : fields) {
            String name = field.getName();
            String value = StrUtil.str(field.get(opentsdbTags), StandardCharsets.UTF_8);
            if (field.getAnnotation(EncodeBase64.class) != null) {
                value = encodeBase64(value);
            }
            mapping.put(name, value);
        }
        return mapping;
    }
    public static void main(String[] args) throws IllegalAccessException {
        OpentsdbTags opentsdbTags = new OpentsdbTags();
        opentsdbTags.setSnCode("587BCL");
        opentsdbTags.setMonitorName("设备实时电压");
        opentsdbTags.setDataUnit("V");
        opentsdbTags.setMonitorCode("0x0501");
        opentsdbTags.setEquipmentTypeCode("ND");
        Number value = 0.005;
        Point.metric("monitor.equipment_szsw").tag(OpenTSDBUtils.encodeToMap(opentsdbTags)).value(1700031167, NumberUtil.round(value.doubleValue(), 3)).build();
    }
    /**
     * 解码为实体
     * @return map
     */
    public static OpentsdbTags decodeToBean(Map<String, String> tags) {
        try {
            OpentsdbTags result = null;
            if (!MapUtil.isEmpty(tags)) {
                result = new OpentsdbTags();
                Field[] declaredFields = result.getClass().getDeclaredFields();
                for (Field field : declaredFields) {
                    String name = field.getName();
                    String value = tags.get(name);
                    if (ObjectUtil.isNotEmpty(value)) {
                        if (field.getAnnotation(EncodeBase64.class) != null) {
                            value = decodeBase64(value);
                        }
                        if (field.getType().isAssignableFrom(value.getClass())) {
                            PropertyDescriptor descriptor = new PropertyDescriptor(field.getName(), result.getClass());
                            Method writeMethod = descriptor.getWriteMethod();
                            writeMethod.invoke(result, value);
                        }
                    }
                }
                // ***** 处理有些历史数据 monitorName 与 dataUnit 存反了的问题 *****
                if (StrUtil.nullToEmpty(result.getDataUnit()).matches("[\u4E00-\u9FA5]+") && !"国标".equals(result.getDataUnit())) {
                    String monitorName = result.getDataUnit();
                    result.setDataUnit(result.getMonitorName());
                    result.setMonitorName(monitorName);
                }
            }
            return result;
        } catch (IllegalAccessException | IntrospectionException | InvocationTargetException e) {
            throw new RuntimeException("创建 OpentsdbTags 对象失败" + e.getMessage());
        }
    }


    /**
     * base64编码
     * @param value 需要编码的值
     * @return 编码后的值
     */
    public static String encodeBase64(String value) {
        return Base64.encodeUrlSafe(value, CharsetUtil.UTF_8);
    }

    /**
     * base64解码
     * @param base64 base64编码的字符串
     * @return 解码后的字符串
     */
    public static String decodeBase64(String base64) {
        return Base64.decodeStr(base64, CharsetUtil.UTF_8);
    }


    /**
     * 判断传入设备snCode样式 创建设备编码过滤器
     * FilterType: IWILDCARD通配符, LITERAL_OR条件过滤
     * tagk: 指标的字段名
     * filter: 与FilterType搭配使用, 根据不同的过滤类型进行不同的过滤
     * groupBy: 是否按照过滤后的指标进行分组?
     */
    public static Filter createSnCodeFilter(String ...snCodes){
        // 不传递则查询所有设备
        Filter snCodeFilter;
        if (ArrayUtil.isEmpty(snCodes)) {
            // 所有设备 使用通配符 *
            snCodeFilter = OpenTSDBUtils.createFilter(
                    FilterType.IWILDCARD,
                    "snCode",
                    "*",
                    true
            );
        } else {
            // 多个设备编码使用 "|" 进行分隔
            String querySnCodeStr = ArrayUtil.join(snCodes, "|");
            snCodeFilter = OpenTSDBUtils.createFilter(
                    FilterType.LITERAL_OR,
                    "snCode",
                    querySnCodeStr,
                    true
            );
        }
        return snCodeFilter;
    }

    /**
     * 判断传入monitorCode样式 创建检测编码过滤器
     * FilterType: IWILDCARD通配符, LITERAL_OR条件过滤
     * tagk: 指标的字段名
     * filter: 与FilterType搭配使用, 根据不同的过滤类型进行不同的过滤
     * groupBy: 是否按照过滤后的指标进行分组?
     */
    public static Filter createMonitorCodeFilter(String ...monitorCodes){
        Filter monitorCodeFilter;
        if (ArrayUtil.isEmpty(monitorCodes)) {
            // 所有指标 使用通配符 *
            monitorCodeFilter = OpenTSDBUtils.createFilter(
                    FilterType.IWILDCARD,
                    "monitorCode",
                    "*",
                    true
            );
        } else {
            // 多个设备编码使用 "|" 进行分隔
            String queryMonitorCodeStr = ArrayUtil.join(monitorCodes, "|");
            monitorCodeFilter = OpenTSDBUtils.createFilter(
                    FilterType.LITERAL_OR,
                    "monitorCode",
                    queryMonitorCodeStr,
                    true
            );
        }
        return monitorCodeFilter;
    }

}
